/**
 * 抖音关注：程序员三丙
 * 知识星球：https://t.zsxq.com/j9b21
 */
package sanbing.example.dylike.infrastructure.gateway;


import com.google.common.util.concurrent.ListenableFuture;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Profile;
import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
import sanbing.example.dylike.domain.LikeCacheGateway;
import sanbing.example.dylike.infrastructure.dataobject.VLikeDO;
import sanbing.example.dylike.infrastructure.sql.dao.LikeDao;
import sanbing.example.dylike.util.MoreThread;

import java.util.Objects;
import java.util.Optional;
import java.util.UUID;

@Component
@Slf4j
@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "redis")
@Profile("!test")
public class RedisLikeCacheGateway implements LikeCacheGateway {

    @Resource
    ReactiveRedisTemplate<UUID, Integer> reactiveRedisTemplate;

    @Resource
    LikeDao likeDao;


    @Override
    public void write(UUID vId, Integer likeNum) {
        reactiveRedisTemplate.opsForValue().set(vId, likeNum, DEFAULT_EXPIRE).block();
    }

    @Override
    public ListenableFuture<Integer> read(UUID vId) {
        return MoreThread.toListenableFuture(reactiveRedisTemplate.opsForValue().get(vId)
                .switchIfEmpty(Mono.defer(() -> Mono.just(Optional.ofNullable(likeDao.get(vId)).map(VLikeDO::getLikeNum).orElse(-1))
                        .flatMap(value -> reactiveRedisTemplate.opsForValue().set(vId, value, DEFAULT_EXPIRE)
                                .thenReturn(value))))
        );
    }

    @Override
    public int total() {
        return getClusterNodesDbSize();
    }

    public int getClusterNodesDbSize() {
        return Objects.requireNonNull(reactiveRedisTemplate.getConnectionFactory().getReactiveClusterConnection()
                .clusterGetNodes()
                .flatMap(node -> getDbSizeForNode(node).map(dbSize -> {
                            log.info("Redis Node: {}, Type: {}, DB Size: {}",
                                    node.asString(),
                                    node.getType(),
                                    dbSize);
                            return dbSize;
                        })
                ).reduce(Long::sum).block()).intValue();
    }

    private Mono<Long> getDbSizeForNode(RedisClusterNode node) {
        return reactiveRedisTemplate.getConnectionFactory().getReactiveClusterConnection()
                .serverCommands()
                .dbSize(node)
                .onErrorResume(e -> Mono.just(-1L));
    }

}